package tableAndSql;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

/**
 * @desc
 * @Author xuyouchang
 * @Date 2021/8/5
 * @Update 2021/8/5 15:41
 */
public class KafkaToEs {
    public static void main(String[] args) {
        String esIndex = "json_10_fields_flink";
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        TableEnvironment bsTableEnv = TableEnvironment.create(bsSettings);



        bsTableEnv.executeSql("create table kafkaTable (`id` BIGINT, `name` VARCHAR, `addr` VARCHAR, `company` VARCHAR, `create_time` BIGINT, `update_time` BIGINT ) " +
                "with ('connector' = 'kafka','topic' = 'json_10_fields','properties.bootstrap.servers' = " +
                "'172.17.1.101:9092','properties.group.id' = 'test10Group','scan.startup.mode' = 'earliest-offset','format' = 'json')");

        bsTableEnv.executeSql("create table esTable (`id` BIGINT, `name` VARCHAR, `addr` VARCHAR, `company` VARCHAR, `create_time` BIGINT, `update_time` BIGINT) " +
                "with ('connector' = 'elasticsearch-7','index' = '"+ esIndex + "','hosts' = " +
                "'http://172.17.1.101:9200')");

        Table table2 = bsTableEnv.sqlQuery("select * from kafkaTable");
        table2.executeInsert("esTable");


    }
}
